use std::{
	net::{IpAddr, SocketAddr, UdpSocket as StdUdpSocket},
	time::Duration,
};

use bytes::{Buf, BytesMut};
use tokio::{
	net::UdpSocket,
	sync::mpsc::{self},
	time::{self},
};
use tools::{
	error::Error, global_defin::tokio_runtime::get_rt, utils::gen_id::get_now_duration, ResultEx,
};
use tracing::{error, info, span, trace, Instrument, Level};

use crate::{
	audio_codes::AudioCodes,
	empty_sockaddr,
	sdp::{
		field::{Media, RtpItem},
		local_sdp,
	},
	sip2::{config::sip_config, dialog::DialogMessage},
	wav::control::{start_run_wav_file_control, WavFileControlEvent},
};

use super::{event::RtpEvent, head::RtpHead, lag::RtpLeg};

///一个rtp声音交换的控制对象.
///
/// 此对象控制声音交换两端..即接收一边的数据并发给另一边.这中间可能产生编码格式的变化.
/// 此对象也控制录音文件的保存.
/// 会启动一个独立的协程用来处理数据的到达与交换,并当两边的发送者都退出,则此对象也退出.
/// 由于此协程是相对于sip的dialog协程存在而存在..因此,也根据dialog的退出而退出..
/// 此交换不根据是否没有真实数据包退出
/// 录音只会在aleg上进行录音,因为aleg是必然存在的
#[derive(Debug)]
pub(crate) struct RtpAuditControl {
	aleg: RtpLeg,
	bleg: Option<RtpLeg>,
	rtp_sock: UdpSocket,
	rtcp_sock: UdpSocket,
	pub(crate) from_sip_dialog: mpsc::Receiver<RtpEvent>,
	pub(crate) audio: Vec<RtpItem>,
	///录音文件的路径,这个是全路径,必须具有权限,可以直接写入的
	pub(crate) send_to_rec: Option<mpsc::UnboundedSender<WavFileControlEvent>>,
	state: RtpControlState,
	pub(crate) ptime: Duration,
	pub(crate) audio_codec: (u8, AudioCodes),
	pub(crate) tele_event_codec: Option<(u8, AudioCodes)>,
}

impl Drop for RtpAuditControl {
	fn drop(&mut self) {
		trace!("rtp 交换关闭.");
		if let Some(rec) = self.send_to_rec.take() {
			let _ = rec.send(WavFileControlEvent::StopRecordFile(self.rec_id()));
		}
	}
}

fn default_codes() -> (u8, AudioCodes) {
	(AudioCodes::G711A.to_wav_code_type(), AudioCodes::G711A)
}

impl RtpAuditControl {
	///使用来录音的id的值....用来表明一个录音唯一值
	#[inline]
	fn rec_id(&self) -> u32 {
		self.aleg.id
	}
	///判断是否完成准备工作 ,可以进入数据交换
	/// 如果已经在交换阶段,返回成功
	#[inline]
	fn is_prepare_complete(&self) -> bool {
		match (&self.state, &self.bleg) {
			(RtpControlState::DataExchange, _) => true,
			(RtpControlState::Prepare, Some(bleg)) => {
				self.aleg.is_prepare_complete() && bleg.is_prepare_complete()
			}
			_ => false,
		}
	}

	pub(crate) fn new(
		dialog_id: u32,
		rtp_sock: UdpSocket,
		rtcp_sock: UdpSocket,
		to_dialog: mpsc::Sender<DialogMessage>,
		from_sip_dialog: mpsc::Receiver<RtpEvent>,
		ptime: u16,
		peer_addr: SocketAddr,
		media: &Vec<Media>,
	) -> Self {
		Self {
			aleg: RtpLeg::new(dialog_id, to_dialog, peer_addr),
			rtp_sock,
			rtcp_sock,
			bleg: None,
			from_sip_dialog,
			audio: media.get(0).unwrap_or(&local_sdp().m[0]).attribs.clone(),
			send_to_rec: None,
			state: RtpControlState::Prepare,
			ptime: Duration::from_millis(ptime as u64),
			audio_codec: default_codes(),
			tele_event_codec: None,
		}
	}

	pub(crate) async fn start_rtp_run(mut self) {
		trace!("rtp 开始运行.当前值:{self:?}");
		//这里的循环是有可能从一个交换阶段转至准备阶段.
		loop {
			trace!("rtp 状态有修改.查看:{:?}", self.state);
			match &self.state {
				RtpControlState::Prepare => self.start_rtp_wait().await,
				RtpControlState::DataExchange => data_exchange(&mut self).await,
				RtpControlState::Close => return,
			}
		}
	}

	fn stop_save_to_file(&mut self) {
		let Some(rec) = self.send_to_rec.take() else {
			trace!("还没有开始录音,不能停止..");
			return;
		};

		let _ = rec.send(WavFileControlEvent::StopRecordFile(self.rec_id()));
	}

	///得到当前音频的编码格式,在整个过程当中可能会发生变化.
	fn load_audio_codec_format(&self) -> (u8, AudioCodes) {
		self
			.audio
			.iter()
			.find(|item| !item.is_tele_event())
			.map(|item| {
				(
					item.payload_type,
					AudioCodes::from_sdp_type(item.payload_type, &item.value),
				)
			})
			.unwrap_or_else(|| default_codes())
	}

	///得到当前事件的编码格式,在整个过程当中可能会发生变化.
	fn load_audio_codec_event(&self) -> Option<(u8, AudioCodes)> {
		self
			.audio
			.iter()
			.find(|item| item.is_tele_event())
			.map(|item| {
				(
					item.payload_type,
					AudioCodes::from_sdp_type(item.payload_type, &item.value),
				)
			})
	}

	///rtp的准备动作.
	/// 此方法是在准备阶段的处理,因为没有bleg,并且还没有确定使用什么样的解码器等.
	/// 因此这里只等待并处理aleg的相关数据,有针对aleg的放音和录音操作,也在这里进行.
	#[inline]
	pub(crate) async fn start_rtp_wait(&mut self) {
		let cap = 320 * 1024;
		let rec_id = self.rec_id();
		let mut buf = BytesMut::with_capacity(cap);
		let mut buf2 = BytesMut::with_capacity(cap / 4);

		let mut send_wav_timestamp = Duration::ZERO;

		//到达下一次发送声音的时间段
		tokio::pin!(let sleep = time::sleep(Duration::from_millis(50)););
		loop {
			// 用来播放文件
			while self.aleg.need_play_buf() && get_now_duration() >= send_wav_timestamp {
				send_wav_timestamp = self
					.aleg
					.try_play_buf(&self.rtp_sock, self.audio_codec, &self.ptime)
					.await;
				// trace!("播放文件内容.下次播放时间:{:?}", send_wav_timestamp);
			}

			if self.aleg.need_play_buf() {
				sleep.set(time::sleep(send_wav_timestamp - get_now_duration()));
			}
			tokio::select! {
			biased;
			event = self.from_sip_dialog.recv() => {
				match event {
					Some(event) => {
						handle_prepare_event(self, event);
						//如果已经准备好,退出准备阶段.
						if self.state == RtpControlState::DataExchange {
							trace!("准备阶段完成,准备进入数据交换阶段");
							return;
						}
					}
					None => {
						info!("所有dialog端已关闭,退出rtp交换");
						self.state = RtpControlState::Close;
						return;
					}
				}
			}
			msg = self.rtp_sock.recv_buf_from(&mut buf) => {
				match msg {
					Ok((len, remote_addr)) if self.aleg.peer_rtp_addr == remote_addr => {
						let data = buf.split_to(len).freeze();

						// trace!("收到rtp包, {}", data.len());
						match load_rtp_body_type(&data, &self.audio) {
							RtpPacket::Audio => {
								if let Some(ref mut rec) = self.send_to_rec {
									let _ = rec.send(WavFileControlEvent::RecordFile(rec_id, true, data));
								} //没有录音要求就跳过
							}
							RtpPacket::Dtmf => if let Err(e) = self.aleg.handle_tel_event(data){
								error!("发送dtmf事件异常:{e}");
							}
							RtpPacket::Unknow | RtpPacket::Video => {}
						};
					}
					//判断是不是从aleg过来的包
					Ok((len, r))   => {
						trace!("收到需要跳过的rtp包....addr:{r}.需要的addr:{}",self.aleg.peer_rtp_addr);
						buf.advance(len);
						// trace!("收到的rtp消息不是从正常路径到达.应该的地址{}..消息地址:{r}",self.aleg.peer_rtp_addr);
					}
					Err(e) => {
						error!("listener_udp()接收数据包异常: {:?}", e);
					}
				}

				//一个rtp包不会大于1500.
				if buf.capacity() < 3 * 1024 {
					trace!("重新得到buf容量");
					buf = BytesMut::with_capacity(cap);
				}
			}
			_msg = self.rtcp_sock.recv_buf_from(&mut buf2) => {

				trace!("收到rtcp包.目前跳过...长度:{}", buf2.len());
				buf2.clear();
				//一个rtp包不会大于1500.
				if buf2.capacity() < 2 * 1024 {
					buf2 = BytesMut::with_capacity(cap / 4);
				}
			}

			//这里就是打开等待,以便于可能存在的播放声音内容数据
			_ = &mut sleep => {
					sleep.set(time::sleep(Duration::from_millis(100)));
				}
			}
		}
	}

	fn set_b_leg(&mut self, dialog_id: u32, to_dialog: mpsc::Sender<DialogMessage>) {
		self.bleg = Some(RtpLeg::new(dialog_id, to_dialog, empty_sockaddr()));
	}

	///保存音频到文件..此处只是一个准备动作.只有在收到rtp的第一个数据包的时候才进行实际的操作
	#[inline]
	fn prepare_save_to_file(&mut self, file_path: String, channel_num: u8) {
		match self.send_to_rec {
			Some(_) => error!("已经存在正在录音的文件,不能再次准备"),
			None => {
				let seder = start_run_wav_file_control();
				let _ = seder.send(WavFileControlEvent::StartRecordFile(
					self.rec_id(),
					file_path,
					AudioCodes::G711A,
					channel_num,
				));
				self.send_to_rec = Some(seder);
			}
		}
	}

	///进入数据交换阶段的一些必要的处理.
	#[inline]
	fn into_data_ex(&mut self) {
		self.state = RtpControlState::DataExchange;
	}
}

///用来处理交换之前的准备阶段接收到的事件.
///
/// 因为可能较多,所以单独分出一个函数来写,也可能不需要异步.写着看
#[inline(always)]
fn handle_prepare_event(slf: &mut RtpAuditControl, event: RtpEvent) {
	trace!("准备阶段事件:{:?}", &event);
	match event {
		RtpEvent::SetBLag(id, to_dialog) => {
			slf.set_b_leg(id, to_dialog);
		}
		RtpEvent::StartRecordFile(file_path) => slf.prepare_save_to_file(file_path, 2),
		RtpEvent::StopRecord => slf.stop_save_to_file(),
		RtpEvent::SetLagAudioMedia(items) => {
			slf.audio = items;
			slf.audio_codec = slf.load_audio_codec_format();
			slf.tele_event_codec = slf.load_audio_codec_event();
		}
		RtpEvent::Bind if slf.is_prepare_complete() => slf.into_data_ex(),
		RtpEvent::Bind => error!("rtp还未准备好..不能进入绑定..状态:{slf:?}"),
		//此事件在当前处理中忽略
		RtpEvent::SetLagVedioMedia(_, _) => trace!("准备阶段忽略视频媒体信息"),
		RtpEvent::SetPeerRtpAddr(id, peer_rtp_addr) if id == slf.aleg.id => {
			slf.aleg.peer_rtp_addr = peer_rtp_addr;
		}
		RtpEvent::SetPeerRtcpAddr(id, peer_rtcp_addr) if id == slf.aleg.id => {
			slf.aleg.peer_rtcp_addr = peer_rtcp_addr;
		}
		RtpEvent::SetPeerRtpAddr(id, peer_rtp_addr)
			if slf.bleg.as_ref().map_or(false, |x| x.id == id) =>
		{
			match slf.bleg {
				Some(ref mut bleg) => bleg.peer_rtp_addr = peer_rtp_addr,
				None => error!("没有bleg.无法设置rtp_addr"),
			}
		}
		RtpEvent::SetPeerRtcpAddr(id, peer_rtcp_addr)
			if slf.bleg.as_ref().map_or(false, |x| x.id == id) =>
		{
			match slf.bleg {
				Some(ref mut bleg) => bleg.peer_rtcp_addr = peer_rtcp_addr,
				None => error!("没有bleg值.无法设置rtcp_addr"),
			}
		}
		RtpEvent::PlayFile(id, wav_buf) if id == slf.aleg.id => {
			slf.aleg.start_play_buf(wav_buf);
		}
		RtpEvent::PlayListFiles(id, bs) if id == slf.aleg.id => {
			slf.aleg.add_play_buf(bs);
		}
		RtpEvent::StopPlay(id) if id == slf.aleg.id => slf.aleg.stop_play(),
		RtpEvent::DialogClose(id) if id == slf.aleg.id => slf.state = RtpControlState::Close,
		_ => error!("准备阶段无效的dialog_id"),
	}
}

///开始数据交换的操作..此时应该已经存在两端,并且已经确定使用什么解码器等.
///
/// 此方法应该在准备阶段结束之后调用.
/// 在循环里面只能使用break.不能直接返回
async fn data_exchange(slf: &mut RtpAuditControl) {
	trace!("进入数据交换阶段");
	let rec_id = slf.rec_id();
	let cap = 1024 * 1024;
	let mut buf = BytesMut::with_capacity(cap);
	let mut o_buf = BytesMut::with_capacity(cap);
	let Some(mut bleg) = slf.bleg.take() else {
		error!("交换阶段必须有bleg端.");
		slf.state = RtpControlState::Prepare;
		return;
	};

	loop {
		tokio::select! {
			biased;
			event = slf.from_sip_dialog.recv() => {
				match event {
					Some(RtpEvent::StartRecordFile(path)) => slf.prepare_save_to_file(path, 2),
					Some(RtpEvent::StopRecord) => slf.stop_save_to_file(),
					Some(RtpEvent::SetLagAudioMedia(items)) => slf.audio = items,
					Some(RtpEvent::SetLagVedioMedia(_, _)) => {
						trace!("当前处理忽略视频媒体信息");
					}
					Some(RtpEvent::SetPeerRtpAddr(id, addr)) if id == slf.aleg.id => {
						if addr != slf.aleg.peer_rtp_addr {
							slf.aleg.peer_rtp_addr = addr;
						}
					}
					Some(RtpEvent::SetPeerRtpAddr(id, addr)) if id == bleg.id => {
						if addr != bleg.peer_rtp_addr {
							bleg.peer_rtp_addr = addr;
						}
					}
					Some(RtpEvent::SetPeerRtcpAddr(id, addr)) if id == slf.aleg.id => {
						if addr != slf.aleg.peer_rtcp_addr {
							slf.aleg.peer_rtcp_addr = addr;
						}
					}
					Some(RtpEvent::SetPeerRtcpAddr(id, addr)) if id == bleg.id => {
						if addr != bleg.peer_rtcp_addr {
							bleg.peer_rtcp_addr = addr;
						}
					}
					Some(RtpEvent::SetBLag(_, _)) => {
						error!("交换过程当中不能设置b端信息");
					}
					Some(other) => error!("没有处理的事件:{:?}", other),
					None => {
						trace!("dialog退出..交换阶段结束");
						slf.state = RtpControlState::Close;
						break;
					}
				}
			}
			msg = slf.rtp_sock.recv_buf_from(&mut buf) => {
				match msg {
				Ok((len, remote_addr))  => {
					let mut data = buf.split_to(len);

					let (to_leg,soure_is_aleg) = if remote_addr == bleg.peer_rtp_addr {
						(&mut slf.aleg,false)
					} else if remote_addr == slf.aleg.peer_rtp_addr {
						(&mut bleg,true)
					} else {
						trace!("收到一个未知来源的数据包.跳过");
						continue;
					};

					match load_rtp_body_type(&data, &slf.audio) {
						RtpPacket::Audio => {
							let _ = to_leg.send_buf_to_rtp(&slf.rtp_sock,slf.audio_codec, &mut data).await;

							if let Some(ref mut rec) = slf.send_to_rec {
								let _ = rec.send(WavFileControlEvent::RecordFile(
									rec_id,
									soure_is_aleg,
									data.freeze(),
								));
							} //没有录音要求就跳过
						}
						RtpPacket::Dtmf => {
							let _ = to_leg.send_tele_event_buf_to_rtp(&slf.rtp_sock,&slf.tele_event_codec, &mut data)
								.await;
						}
						RtpPacket::Unknow | RtpPacket::Video => {}
					};
				}
				Err(e) => {
					error!("listener_udp()接收数据包异常: {:?}", e);
				}
				}
			}
			msg = slf.rtcp_sock.recv_buf_from(&mut o_buf) => {
				match msg {
					Ok((len, remote_addr)) => {
						let data = o_buf.split_to(len).freeze();
						let _ = if remote_addr == slf.aleg.peer_rtp_addr {
							&mut bleg
						} else if remote_addr == bleg.peer_rtp_addr {
							&mut slf.aleg
						} else {
							trace!("收到一个未知来源的数据包.跳过");
							continue;
						};

						trace!("bleg收到rtcp 消息:{:?}", data);
					}
					// Ok((len, _)) => {
					// 	let _ = buf.split_to(len).freeze();
					// 	trace!("bleg收到的rtcp消息不是正确的地址发来的消息.")
					// }
					Err(e) => error!("bleg,rtp接收异常:{e}"),
				}
			}
		}
	}

	slf.bleg = Some(bleg);
}

///当前rtp管理端的状态
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
enum RtpControlState {
	Prepare,
	DataExchange,
	Close,
}

/// 判断此包是否在指定的几种包协议中, 返回对应的包的类型.
///
/// 因为中转两边都需要同一个解码格式,因此只进行确认当前发来的数据可被编码,而不进行转码动作
fn load_rtp_body_type(buf: &[u8], rtp_items: &Vec<RtpItem>) -> RtpPacket {
	let pt = RtpHead::load_payload_type(buf);

	for item in rtp_items.iter() {
		if item.payload_type == pt {
			if item.is_tele_event() {
				return RtpPacket::Dtmf;
			} else {
				return RtpPacket::Audio;
			}
		}
	}

	trace!("没有找到对应的类型..buf_wav_type:{pt},,rtp_items:{rtp_items:?}");
	return RtpPacket::Unknow;
}

///此类是从rtp包当中提取出来的格式和id
/// 因为部分id是动态的,所以需要在这个类里标识.
/// 一般用来于sdp的信息进行匹配
enum RtpPacket {
	Audio,
	///为以后保留
	#[allow(unused)]
	Video,
	Dtmf,
	Unknow,
}

pub(crate) fn start_rtp_run_control(
	id: u32,
	rtp_to_dialog: mpsc::Sender<DialogMessage>,
	peer_addr: SocketAddr,
	rtp_sock: UdpSocket,
	rtcp_sock: UdpSocket,
	medias: &Vec<Media>,
) -> ResultEx<mpsc::Sender<RtpEvent>> {
	let (to_rtp, rtp_rev) = mpsc::channel(256);
	//这里现在只启动声音的交换
	let rtp_m = RtpAuditControl::new(
		id,
		rtp_sock,
		rtcp_sock,
		rtp_to_dialog,
		rtp_rev,
		sip_config().ptime,
		peer_addr,
		medias,
	);
	let rtp = id;
	let span = span!(Level::ERROR, "", rtp);
	get_rt().spawn(rtp_m.start_rtp_run().instrument(span));

	Ok(to_rtp)
}

///返回指定ip的监听对象
///
/// 因为在使用端口时直接保证端口被当前使用.
pub(crate) fn generate_rtp_listen(ip: IpAddr) -> ResultEx<(UdpSocket, UdpSocket)> {
	for _i in 0..100 {
		let rtp_sock = match StdUdpSocket::bind((ip, 0)) {
			Ok(s) => s,
			Err(e) => {
				error!("绑定异常: {:?}", e);
				return Err(e.into());
			}
		};
		let port = rtp_sock.local_addr()?.port();
		let rtpc_sock = match StdUdpSocket::bind((ip, port + 1)) {
			Ok(s) => s,
			Err(_) => {
				trace!("连续端口被占用.");
				continue;
			}
		};

		trace!(
			"得到两个连续的端口:{:?},{:?}",
			rtp_sock.local_addr(),
			rtpc_sock.local_addr()
		);

		rtp_sock.set_nonblocking(true)?;
		rtpc_sock.set_nonblocking(true)?;
		return Ok((
			UdpSocket::from_std(rtp_sock)?,
			UdpSocket::from_std(rtpc_sock)?,
		));
	}

	Err(Error::IOError(
		"连续100次没有拿到两个连续的空闲端口".to_owned(),
	))
}
